/*
 * @Author: wuqingchun
 * @Date: 2024-09-23 14:42:02
 * @Description: 
 * @LastEditors: wuqingchun
 * @LastEditTime: 2024-10-03 21:23:13
 */

#include "../chat_errno.h"
#include "chat_redis_subpub.h"

CChatRedisSubPub::CChatRedisSubPub()
{

}

CChatRedisSubPub::~CChatRedisSubPub()
{
    
}

int32_t CChatRedisSubPub::Init()
{
    m_pstEventBase = event_base_new();
    if (m_pstEventBase == nullptr)
    {
        return CHAT_ERR_ALLOC;
    }

    return CHAT_ERR_OK;
}

void CChatRedisSubPub::Exit()
{
    if (m_pstRedisCtx != nullptr)
    {
        redisAsyncFree(m_pstRedisCtx);
        m_pstRedisCtx = nullptr;
    }

    if (m_pstEventBase != nullptr)
    {
        m_pstEventBase = nullptr;
    }
}

int32_t CChatRedisSubPub::Connect(const char* pcHost, uint16_t usPort)
{
    if (m_pstEventBase == nullptr)
    {
        printf("No initialization\n");
        return CHAT_ERR_INIT;
    }

    m_pstRedisCtx = redisAsyncConnect(pcHost, usPort);
    if (m_pstRedisCtx == nullptr)
    {
        printf("Failed to connect redis, host=%s, port=%hu\n", pcHost, usPort);
        return CHAT_ERR_CONNECT_REDIS;
    }

    if (m_pstRedisCtx->err != 0)
    {
        printf("Failed to connect redis, host=%s, port=%hu, err=%d, errstr=%s\n", pcHost, usPort, 
               m_pstRedisCtx->err, m_pstRedisCtx->errstr);
        return CHAT_ERR_CONNECT_REDIS;
    }

    // 将事件绑定到redis context上，使设置给redis的回调跟事件关联
    int iErr = redisLibeventAttach(m_pstRedisCtx, m_pstEventBase);
    if (iErr != REDIS_OK)
    {
        return CHAT_ERR_CONNECT_REDIS;
    }

    (void)redisAsyncSetConnectCallback(m_pstRedisCtx, &CChatRedisSubPub::__OnConnect);
    (void)redisAsyncSetDisconnectCallback(m_pstRedisCtx, &CChatRedisSubPub::__OnDisconnect);

    m_objEventThread = std::thread(&CChatRedisSubPub::__ThreadEventFunc, this);

    return CHAT_ERR_OK;
}

int32_t CChatRedisSubPub::Publish(int iChannel, const char* pcMessage)
{
    if (m_pstEventBase == nullptr || m_pstRedisCtx == nullptr)
    {
        return CHAT_ERR_INIT;
    }
    
    int ret = redisAsyncCommand(m_pstRedisCtx, &CChatRedisSubPub::__OnPublishReply, this, "PUBLISH %d %s", 
                                iChannel, pcMessage);
    if (REDIS_ERR == ret)
    {
        printf("Publish command failed: %d\n", ret);
        return CHAT_ERR_REDIS_PUBLISH;
    }
 
    return CHAT_ERR_OK;
}

int32_t CChatRedisSubPub::Publish(const char* pcChannel, const char* pcMessage)
{
    if (m_pstEventBase == nullptr || m_pstRedisCtx == nullptr)
    {
        return CHAT_ERR_INIT;
    }
    
    int ret = redisAsyncCommand(m_pstRedisCtx, &CChatRedisSubPub::__OnPublishReply, this, "PUBLISH %s %s", 
                                pcChannel, pcMessage);
    if (REDIS_ERR == ret)
    {
        printf("Publish command failed: %d\n", ret);
        return CHAT_ERR_REDIS_PUBLISH;
    }
 
    return CHAT_ERR_OK;
}

int32_t CChatRedisSubPub::Subscribe(int iChannel)
{
    if (m_pstEventBase == nullptr || m_pstRedisCtx == nullptr)
    {
        return CHAT_ERR_INIT;
    }
    
    printf("SUBSCRIBE %d\n", iChannel);
    int ret = redisAsyncCommand(m_pstRedisCtx, &CChatRedisSubPub::__OnSubscribeReply, this, "SUBSCRIBE %d", 
                                iChannel);
    if (REDIS_ERR == ret)
    {
        printf("Publish command failed: %d\n", ret);
        return CHAT_ERR_REDIS_PUBLISH;
    }
 
    return CHAT_ERR_OK;
}

int32_t CChatRedisSubPub::Subscribe(const char* pcChannel)
{
    if (m_pstEventBase == nullptr || m_pstRedisCtx == nullptr)
    {
        return CHAT_ERR_INIT;
    }
    
    int ret = redisAsyncCommand(m_pstRedisCtx, &CChatRedisSubPub::__OnSubscribeReply, this, "SUBSCRIBE %s", 
                                pcChannel);
    if (REDIS_ERR == ret)
    {
        printf("Publish command failed: %d\n", ret);
        return CHAT_ERR_REDIS_PUBLISH;
    }
 
    return CHAT_ERR_OK;
}


void CChatRedisSubPub::__OnConnect(const redisAsyncContext* pstRedisCtx, int iStatus)
{
    if (iStatus != REDIS_OK)
    {
        printf("Failed to connect redis, %s\n", pstRedisCtx->errstr);
    }
    else
    {
        printf("Connected redis successfully\n");
    }
}

void CChatRedisSubPub::__OnDisconnect(const redisAsyncContext* pstRedisCtx, int iStatus)
{
    if (iStatus != REDIS_OK)
    {
        // 这里异常退出，可以尝试重连
        printf(": Error: %s\n", pstRedisCtx->errstr);
    }
}

void CChatRedisSubPub::__OnPublishReply(struct redisAsyncContext* pstRedisCtx, void* pReply, void* pPrivate)
{
    printf("Received publish reply\n");
}

void CChatRedisSubPub::__OnSubscribeReply(struct redisAsyncContext* pstRedisCtx, void* pReply, void* pPrivate)
{
    printf("Received subscribe message\n");

    if (NULL == pReply || NULL == pPrivate) {
        return ;
    }
 
    // 静态函数中，要使用类的成员变量，把当前的this指针传进来，用this指针间接访问
    CChatRedisSubPub *self_this = reinterpret_cast<CChatRedisSubPub *>(pPrivate);
    redisReply *pstReply = reinterpret_cast<redisReply *>(pReply);
 
    // 订阅接收到的消息是一个带三元素的数组
    if (pstReply->type == REDIS_REPLY_ARRAY && pstReply->elements == 3)
    {
        printf("Recieved message:%s:%lu:%s:%lu:%s:%lu\n",
        pstReply->element[0]->str, pstReply->element[0]->len,
        pstReply->element[1]->str, pstReply->element[1]->len,
        pstReply->element[2]->str, pstReply->element[2]->len);
 
        // 调用函数对象把消息通知给外层
        // self_this->_notify_message_fn(redis_reply->element[1]->str,
        //     redis_reply->element[2]->str, redis_reply->element[2]->len);
    }
}

void CChatRedisSubPub::__ThreadEventFunc()
{
    // 开启事件分发，event_base_dispatch会阻塞
    event_base_dispatch(m_pstEventBase);
}

